package org.jmqtt.broker.store.redis.support;

import org.jmqtt.broker.common.config.BrokerConfig;
import org.jmqtt.broker.common.log.JmqttLogger;
import org.jmqtt.broker.common.log.LogUtil;
import org.slf4j.Logger;

import java.lang.reflect.Constructor;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * redis 访问模板方法类
 */
public class RedisUtils {

    private static final Logger log = JmqttLogger.otherLog;

    private static final RedisUtils redisUtils = new RedisUtils();

    private volatile RedisSupport redisSupport;

    private AtomicBoolean start = new AtomicBoolean(false);

    private RedisUtils(){}

    public static RedisUtils getInstance(){
        return redisUtils;
    }

    public RedisSupport createSupport(BrokerConfig brokerConfig) {
        if (start.compareAndSet(false, true)) {
            // this.redisSupport = new RedisSupportImpl(brokerConfig);
            try {
                // 使用配置反射实例化redis
                Class<?> redisSupportClass = Class.forName(brokerConfig.getRedisSupportClass());
                Constructor<?> redisSupportConstructor = redisSupportClass.getConstructor(BrokerConfig.class);
                this.redisSupport = (RedisSupport) redisSupportConstructor.newInstance(brokerConfig);
            } catch (Exception e) {
                LogUtil.error(log, "init redis error, ex:{}", e);
            }
            this.redisSupport.init();
        }
        return redisSupport;
    }

    public void close() {
        RedisSupport redisSupport = this.redisSupport;
        if(start.compareAndSet(true,false)&&redisSupport!=null){
            redisSupport.close();
        }
    }
}
